package com.chb.catRecommend;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import sun.misc.Perf;


public class Step4 {
	public static boolean run(Configuration conf, Map<String, String> paths) throws Exception{
		FileSystem fs = FileSystem.get(conf);
		Job job = Job.getInstance();
		job.setJar("C:\\Users\\12285\\Desktop\\cr.jar");
		job.setJarByClass(Step4.class);
		job.setJobName("Step4");

		job.setMapperClass(Step4Mapper.class);
		job.setReducerClass(Step4Reducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		// org.apache.hadoop.io.LongWritable cannot be cast to org.apache.hadoop.io.Text
		job.setInputFormatClass(KeyValueTextInputFormat.class);
			
		//将喜爱得分的表加载到内存中。
		job.addCacheFile(new Path(paths.get("step4InputSmall")).toUri());

		Path in = new Path(paths.get("step4InputBig"));
		FileInputFormat.addInputPath(job, in);
		Path out = new Path(paths.get("step4Output"));
		if (fs.exists(out)) {
			fs.delete(out, true);
		}
		
		FileOutputFormat.setOutputPath(job, out);
		boolean f = job.waitForCompletion(true);
		return f;

	} 


	static class Step4Mapper extends Mapper<Text, Text, Text, IntWritable> {
		Map<String, Map<String, Integer>> prefMap = new HashMap<String, Map<String, Integer>>();
		@Override
		protected void setup(Context context)
				throws IOException, InterruptedException {
			// user_id itemId:得分 。。。
			Path path = new Path(context.getCacheFiles()[0].getPath());
			//读取信息:u26	i276:1,i201:1,i348:1,i321:1,i136:1,
			BufferedReader br = new BufferedReader(new  FileReader(path.getName()));
			String line = null;
			while((line = br.readLine()) != null) {
				String user_id = line.split("\t")[0];
				String[] itemIds_score = line.split("\t")[1].split(",");
				Map<String, Integer> maps = new HashMap<String, Integer>();
				for (String itemsc : itemIds_score) {
					if (itemsc != null) {
						String item = itemsc.split(":")[0];
						int score = Integer.parseInt(itemsc.split(":")[1]);
						maps.put(item, score);
					}
				}
				//{userID: {itemID:喜爱评分}}
				prefMap.put(user_id, maps);
			}
		}
		@Override
		protected void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			/**
			 i100:i184	2
			 i100:i185	1
			 i100:i187	1
			 */
//			Map<String, Integer> txMarix  = new HashMap<String, Integer>();
//			txMarix.put(key.toString(), Integer.parseInt(value.toString()));

			//{userID: {itemID:喜爱评分}}
			Set<String> user_ids = prefMap.keySet();
			for (String userId : user_ids) {
			    Map<String, Integer> itemMap = prefMap.get(userId);
			    
			    //同现矩阵，i100:i184	2
			    String itemA = key.toString().split(":")[0];
			    String itemB = key.toString().split(":")[1];
			    int txNum = Integer.parseInt(value.toString());
			    int  prefScore = 0;
			   //获取用户对itemB的喜爱得分
			    if (itemMap.get(itemB) != null) {//Error: java.lang.NullPointerException
			    	prefScore = itemMap.get(itemB);
				}
					   
			   //推荐向量的分量：itemB贡献 的为itemA和itemB同现次数*用户对ItemB的喜爱评分
			   int uItemB = txNum * prefScore;
				//输出
			   context.write(new Text(userId+":"+itemA), new IntWritable(uItemB));
			}
		}

	}
	
	/**
	 *  
	 */
	static class Step4Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable iw : values) {
				sum += iw.get();
			}
			context.write(key, new IntWritable(sum));
		}
	}
}
